package com.thp.bigdata.wcdemo;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;


/**
 * 驱动类
 * 相当于一个yarn集群的客户端
 * 需要在此封装mapreduce程序的相关运行参数,指定jar包
 * 最后提交给yarn
 * @author 汤小萌
 *
 */
public class WordcountDriver {
	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
		Configuration conf = new Configuration();
		
		// conf.set("fs.hdfs.impl",org.apache.hadoop.hdfs.DistributedFileSystem.class.getName());

		
		System.err.println(args[0]);
		System.err.println(args[1]);
		
		String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
		if(otherArgs.length != 2) {
			System.err.println("Usage : wordcount <in> <out>");
			System.exit(2);
		}
		
		
		
		
		/*conf.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");
	    conf.set("mapreduce.framework.name", "yarn");
		conf.set("yarn.resourcemanager.hostname", "bd1");
		
		conf.set("fs.defaultFS", "hdfs://bd1:9000");*/
		
		//是否运行为本地模式，就是看这个参数值是否为local，默认就是local
		/*conf.set("mapreduce.framework.name", "local");*/
		
		//本地模式运行mr程序时，输入输出的数据可以在本地，也可以在hdfs上
		//到底在哪里，就看以下两行配置你用哪行，默认就是file:///
		/*conf.set("fs.defaultFS", "hdfs://mini1:9000/");*/
		/*conf.set("fs.defaultFS", "file:///");*/
		
		Job job = Job.getInstance(conf);
		
		// 指定本程序的jar包所在的本地路径
		/*job.setJarByClass(WordcountDriver.class);*/
		job.setJar("f:/mr.jar");
		
		// 指定本业务job要使用的mapper/Reduce业务类
		job.setMapperClass(WordcountMapper.class);
		job.setReducerClass(WordcountReduce.class);
		// 设置Map规约Combiner
		job.setCombinerClass(MyCombiner.class);
		
		// 指定mapper输出数据的kv类型
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(IntWritable.class);
		
		// 指定最终输出的数据的kv类型
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		
		
		
		Path path = new Path(otherArgs[1]);
		FileSystem fileSystem = path.getFileSystem(conf);   // 根据path找到这个文件
		if(fileSystem.exists(path)) {
			fileSystem.delete(path, true);  // true的意思是，就算output有东西，也一带删除
		}
		
		
		// 指定job的输入原始文件所在的目录
		// 待处理文件可以在多个目录里面
		FileInputFormat.setInputPaths(job, new Path(args[0]));
		
		// 指定job的输出结果
		FileOutputFormat.setOutputPath(job, new Path(args[1]));
		
		
		
		// 将job中配置的相关参数,以及job所用的的java类所在的jar包,提交给yarn去运行
		/*job.submit();*/
		boolean res = job.waitForCompletion(true); // 会等待程序处理完成之后,程序才退出
		System.exit(res ? 0 : 1);
		
	}
}
